package com.bff.gaia.streaming.api.PLO;

import com.bff.gaia.streaming.api.functions.AssignerWithPeriodicWatermarks;
import com.bff.gaia.streaming.api.watermark.Watermark;

public class PLO implements AssignerWithPeriodicWatermarks<Word> {
	long maxTimestamp = 0L;         //最大时间戳
	long shuchushijian = 0L;        //存储输处最大水位线：保证正序
	long delay;                     //延迟时长
	Long[] zuijinshuju = {0L,0L,0L,0L,0L};  //微簇
	double luanxvdu = 1.0;                  //当前乱序度
	Long MaxDD;
	Long sleep = 1L;

	public  PLO(long delay, Long sleep){
		this.MaxDD = (delay)*1000;
		this.sleep = sleep;
	}

	public PLO(long delay){
		this.MaxDD = (delay)*1000;
	}


	public long extractTimestamp(Word event, long l) {
		try {
			//放慢处理速度，否则可能只会生成一条水位线
			Thread.sleep(sleep);
		}
		catch (Exception ex){
		}
		// maxTimestamp = Math.max(event.eventTime, maxTimestamp);
		maxTimestamp = event.eventTime;
		//计算乱序度
		this.zuijinshuju =charu(zuijinshuju,event.eventTime);
		this.luanxvdu = jisuanluanxvdu(zuijinshuju);
		//更新等待时间
		this.delay = (long) (this.MaxDD/2*1*this.luanxvdu+(this.MaxDD/2)*1);
		//this.delay = (long) (this.MaxDD*this.luanxvdu);
		this.shuchushijian =Math.max(shuchushijian, maxTimestamp - delay);
		//提取时间戳
		return event.eventTime;
	}

	public Watermark getCurrentWatermark() {
		return new Watermark(shuchushijian);
	}
	//微簇更新
	public  Long[] charu( Long[] zuijinshuju , Long a){
		for(int i = 0;i < 4;i++){
			zuijinshuju[i] = zuijinshuju[i+1];
		}
		zuijinshuju[4] = a;
		return zuijinshuju;
	}
	//乱序度算法
	double jisuanluanxvdu(Long[] zuijinshuju){
		double fenzi = 0;
		double fenmu = 0;
		for(int i=1;i<5;i++){
			if(zuijinshuju[i]-zuijinshuju[i-1] > 0)
				fenzi+=zuijinshuju[i]-zuijinshuju[i-1];
			fenmu+=zuijinshuju[i]-zuijinshuju[i-1]<0?-(zuijinshuju[i]-zuijinshuju[i-1]):(zuijinshuju[i]-zuijinshuju[i-1]);
		}
		//两种乱序度计算方式
		//return tanh((fenzi/fenmu)*2.7);
		return fenzi/fenmu;
	}
	double tanh(double x) {
		return (Math.pow(Math.E, x) - Math.pow(Math.E, -x)) / (Math.pow(Math.E, x) + Math.pow(Math.E, -x));
	}
}